import json
import time

from kafka import KafkaProducer
from itemadapter import ItemAdapter

class EastmoneyPipeline:
    def __init__(self):
        self.producer = KafkaProducer(
            bootstrap_servers=['192.168.220.131:9092', '192.168.220.132:9092', '192.168.220.133:9092', '192.168.220.134:9092'],
            value_serializer=lambda x: json.dumps(x, ensure_ascii=False).encode('utf-8')
        )
        self.topic_name = 'bigdata'
        self.buffer = []  # 缓存要发送的消息
        self.max_buffer_size = 4  # 每次发送的最大消息数
        self.ready_to_send = True  # 控制是否允许发送新消息的标志

    def open_spider(self, spider):
        pass

    def close_spider(self, spider):
        if self.buffer:
            self._flush_buffer()  # 发送剩余未发送的消息
        self.producer.flush()
        self.producer.close()

    def process_item(self, item, spider):
        try:
            if not self.ready_to_send:
                self.buffer.append(item)
                if len(self.buffer) >= self.max_buffer_size:
                    self._flush_buffer()
            else:
                future = self.producer.send(self.topic_name, value=item)
                # time.sleep(1)
                future.add_callback(self._on_send_success)
                future.add_errback(self._on_send_error)
                print(f"Sent item to Kafka topic {self.topic_name}: {item}")
        except Exception as e:
            spider.logger.error(f"Failed to send item to Kafka: {str(e)}")
        return item

    def _flush_buffer(self):
        """将缓存中的消息批量发送"""
        while self.buffer and self.ready_to_send:
            item = self.buffer.pop(0)
            future = self.producer.send(self.topic_name, value=item)
            future.add_callback(self._on_send_success)
            future.add_errback(self._on_send_error)
            print(f"Flushed item from buffer to Kafka topic {self.topic_name}: {item}")

    def _on_send_success(self, record_metadata):
        """成功发送消息后的回调"""
        print(f"Message sent successfully to partition [{record_metadata.partition}] offset [{record_metadata.offset}]")
        if not self.buffer:
            self.ready_to_send = True  # 如果缓存为空，则准备好发送新消息

    def _on_send_error(self, exc_info):
        """发送消息失败后的回调"""
        print(f"Failed to send message: {exc_info}")
        self.ready_to_send = False  # 发生错误时暂停发送新消息

    def signal_consumption_complete(self):
        """模拟接收到消费者已处理完消息的通知"""
        print("Received signal that all messages have been consumed.")
        self.ready_to_send = True  # 允许再次发送新消息
        if self.buffer:
            self._flush_buffer()  # 立即尝试发送缓存中的消息